package beanstao.util.socket;

import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import beanstao.util.exception.technical.util.SocketDoorException;
import beanstao.util.log.ILog;
import beanstao.util.log.LogHelper;
import beanstao.util.socket.protocol.IProtocolFactory;

/**
 * 一个处理已知协议SocketServer。
 * @author beanstao@google.com 
 */
public class SocketServer extends Thread implements ILog
{
	/**
	 * 描述队列成功状态
	 */
	private static final Object SUCCESS = new Object();
	private final BlockingQueue<Object> startedStatusQueue = new LinkedBlockingQueue<Object>();

	private final ThreadPoolExecutor threadPool;
	private final int port;
	private final ThreadGroup threadGroup;
	private final int socketBufferSize;
	private final IProtocolFactory handlerFactory;
	private final int maxThreads;
	private final StatusManager statusManager;
	private final AtomicLong sessionIdSequence;
	private final ConcurrentMap<Long, SocketServerSession> activeSessions;
	private final String serverName;
	private final boolean isDecodeHead;

	private ServerSocket serverSocket = null;

	public SocketServer(final int port, final int defaultThreads, final int maxThreads, final int receiveBuffer, final IProtocolFactory handlerFactory, final boolean isDecodeHead, final String serverName)
	{
		this.port = port;
		this.handlerFactory = handlerFactory;
		this.maxThreads = maxThreads;
		this.serverName = serverName;
		this.isDecodeHead = isDecodeHead;

		socketBufferSize = receiveBuffer;
		threadGroup = new ThreadGroup("ecms-socket-server-door");
		threadPool = new ThreadPoolExecutor(defaultThreads, maxThreads, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), threadFactory, rejectedExecutionHandler);
		statusManager = new StatusManager(threadPool);
		sessionIdSequence = new AtomicLong(0);
		activeSessions = new ConcurrentHashMap<Long, SocketServerSession>();
	}

	private final ThreadFactory threadFactory = new ThreadFactory()
	{
		private final AtomicLong threadIdSequence = new AtomicLong(0);

		public Thread newThread(final Runnable r)
		{
			String name = "ecms-server-" + threadIdSequence.getAndIncrement();
			Thread t = new Thread(threadGroup, r, name);
			t.setDaemon(true);
			return t;
		}
	};

	private final RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler()
	{
		public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor)
		{
			SocketServerSession session = (SocketServerSession) r;
			if (interrupted())
				LogHelper.logInfo(session, "拒绝连接: " + session.getSocket().getRemoteSocketAddress() + ", 服务关闭.");
			else
				LogHelper.logError(session, "连接数过多, " + executor.getActiveCount() + " of " + executor.getLargestPoolSize() + " 线程已开启。拒绝来自下述的连接: " + session.getSocket().getRemoteSocketAddress());
			try
			{
				session.getSocket().close();
			}
			catch (IOException e)
			{
				LogHelper.logError(session, "异常。未能关闭Socket。", e);
			}
		}
	};

	@Override
	public void run()
	{
		String info = String.format("启动:socket-server监听。服务名:%s。端口号:%s。当前连接(线程)数:%s。最大连接(线程)数:%S。缓冲区大小:%s。", serverName, port, threadPool.getTaskCount(), maxThreads, socketBufferSize);
		this.logInfo(info);
		try
		{
			serverSocket = new ServerSocket();
			serverSocket.bind(new InetSocketAddress(port));
			serverSocket.setReceiveBufferSize(socketBufferSize);
			startedStatusQueue.put(SUCCESS);
			while (!this.isInterrupted() && !serverSocket.isClosed())
			{
				final Socket socket = serverSocket.accept();
				this.configureSocket(socket);
				long sessionId = sessionIdSequence.getAndIncrement();
				threadPool.execute(new SocketServerSession(activeSessions, socket, handlerFactory, isDecodeHead, sessionId));
			}
		}
		catch (BindException e)
		{
			this.logError("不能打开Socket端口。端口号: " + port + ".");
			startedStatusQueue.offer(e);
			throw new SocketDoorException(e);
		}
		catch (SocketException e)
		{
			startedStatusQueue.offer(e);
			if (!this.isInterrupted())
				this.logWarn("绕过异常。系统异常: ", e);
		}
		catch (IOException e)
		{
			startedStatusQueue.offer(e);
			throw new SocketDoorException(e);
		}
		catch (Throwable t)
		{
			this.logError(t);
			startedStatusQueue.offer(t);
			if (t instanceof Error)
				throw (Error) t;
			else if (t instanceof RuntimeException)
				throw (RuntimeException) t;
			throw new SocketDoorException(t);
		}
		finally
		{
			if (serverSocket != null)
				try
				{
					serverSocket.close();
				}
				catch (IOException e)
				{
					this.logWarn("关闭SocketServer时异常.", e);
				}

		}
	}

	private void configureSocket(final Socket socket) throws SocketException
	{
		socket.setTcpNoDelay(true);
		socket.setSendBufferSize(socketBufferSize);
		if (socket.getReceiveBufferSize() != socketBufferSize)
			this.logDebug("接收缓冲区需要 " + socketBufferSize + " bytes, 但是实际是 " + socket.getReceiveBufferSize() + " bytes.");
		if (socket.getSendBufferSize() != socketBufferSize)
			this.logDebug("发送缓冲区需要 " + socketBufferSize + " bytes, 但是实际是 " + socket.getSendBufferSize() + " bytes.");
	}

	/**
	 * 本类型的关闭主方法
	 */
	public void shutdown()
	{
		this.logInfo("SocketServer Shutting down! (" + serverName + ").");

		// 首先，关闭SocketServer，以便停止新的连接
		this.interrupt();
		try
		{
			if (!serverSocket.isClosed())
				serverSocket.close();
		}
		catch (IOException e)
		{
			this.logError("关闭SocketServer时异常: " + e.getMessage());
		}

		// 关闭活动的Session
		threadPool.shutdownNow();
		this.killActiveSessions();

		try
		{
			boolean completed = threadPool.awaitTermination(5, TimeUnit.SECONDS);
			if (!completed)
				this.logWarn("等待线程池关闭超时.");
		}
		catch (InterruptedException e)
		{
			this.logWarn("过程中断。异常。当关闭SocketServer时: ", e);
		}
	}

	/**
	 * 遍历并关闭所有活动的Session
	 */
	public void killActiveSessions()
	{
		this.logInfo("遍历并关闭所有活动的Session.");
		for (Map.Entry<Long, SocketServerSession> entry : activeSessions.entrySet())
			try
			{
				this.logDebug("被关闭的Session: " + entry.getKey());
				entry.getValue().close();
			}
			catch (IOException e)
			{
				this.logWarn("关闭Session时发生异常。 ", e);
			}
	}

	public int getPort()
	{
		return port;
	}

	public int getMaxThreads()
	{
		return maxThreads;
	}

	public int getCurrentThreads()
	{
		return threadPool.getActiveCount();
	}

	public int getRemainingThreads()
	{
		return this.getMaxThreads() - this.getCurrentThreads();
	}

	/**
	 * 阻塞，直到服务器已成功启动。
	 * @throws SocketDoorException
	 */
	public void awaitStartupCompletion()
	{
		try
		{
			Object obj = startedStatusQueue.take();
			if (obj instanceof Throwable)
				throw new SocketDoorException((Throwable) obj);
		}
		catch (InterruptedException e)
		{
			// 这时并非是异常
		}
	}

	public StatusManager getStatusManager()
	{
		return statusManager;
	}

	// = 日志接口(ILog)的实现 ============

	@Override
	public void logDebug(final Object message)
	{
		LogHelper.logDebug(this, message);
	}

	@Override
	public void logDebug(final Object message, final Throwable t)
	{
		LogHelper.logDebug(this, message, t);
	}

	@Override
	public void logError(final Object message)
	{
		LogHelper.logError(this, message);
	}

	@Override
	public void logError(final Object message, final Throwable t)
	{
		LogHelper.logError(this, message, t);
	}

	@Override
	public void logInfo(final Object message)
	{
		LogHelper.logInfo(this, message);
	}

	@Override
	public void logInfo(final Object message, final Throwable t)
	{
		LogHelper.logInfo(this, message, t);
	}

	@Override
	public void logWarn(final Object message)
	{
		LogHelper.logWarn(this, message);
	}

	@Override
	public void logWarn(final Object message, final Throwable t)
	{
		LogHelper.logWarn(this, message, t);
	}

}
